package com.group2.edu.realtime.common.util;

import com.group2.edu.realtime.common.constant.Constant;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.kafka.clients.consumer.OffsetResetStrategy;

import java.io.IOException;
import java.util.Properties;

/**
 * @author 高耀
 * @date 2024/12/14 10:09<p></p>
 *
 * 工具类，用于提供各种FlinkSource
 */
public class FlinkSourceUtil {

    /**
     * 获取KafkaSource的方法
     * @param topic topic主题
     * @param groupId 消费者组名
     */
    public static KafkaSource<String> getKafkaSource(String topic, String groupId) {
        return KafkaSource.<String>builder()
                .setBootstrapServers(Constant.KAFKA_BROKERS_ID)
                .setTopics(topic)
                .setGroupId(groupId)
                .setStartingOffsets(//读取最近提交的Offset，如果没有，就从最新位置读
                        OffsetsInitializer.committedOffsets(OffsetResetStrategy.LATEST))
                .setValueOnlyDeserializer(
                        //自定义value的反序列化器
                        new DeserializationSchema<String>() {
                            /**
                             * 具体反序列化方法
                             * @param bytes 传入的每条需要被序列化的数据
                             * @return 处理结果
                             */
                            @Override
                            public String deserialize(byte[] bytes) throws IOException {
                                //实现判空操作，如果从kafka读到了空值，默认的String反序列化器SimpleStringSchema会报异常
                                if (bytes != null) {
                                    return new String(bytes);
                                } else {
                                    return null;
                                }

                            }

                            @Override
                            public boolean isEndOfStream(String s) {
                                return false;
                            }

                            @Override
                            public TypeInformation<String> getProducedType() {
                                return TypeInformation.of(String.class);
                            }
                        }
                )
                .build();
    }



    /**
     * 使用FlinkCDC获取mysql表的数据
     * @param tableName 表名
     */
    public static MySqlSource<String> getCDCMysqlSource(String tableName) {
        Properties props = new Properties(); //数据库另外的配置项
        props.setProperty("useSSL", "false");
        props.setProperty("allowPublicKeyRetrieval", "true");

        return MySqlSource.<String>builder()
                .hostname(Constant.MYSQL_HOST)
                .username(Constant.MYSQL_USER_NAME)
                .password(Constant.MYSQL_PASSWORD)
                .port(Constant.MYSQL_PORT)
                .deserializer(new JsonDebeziumDeserializationSchema()) //反序列化器，直接转成JSON
                .databaseList("gmall_config") //配置来源数据库
                .tableList("gmall_config." + tableName) //配置来源表
                .jdbcProperties(props)
                .build();
    }
}
